﻿using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Windows.Forms;
using Sinacor.OMM.ConnectionManager.MSMQ;
using System.Threading;
using Sinacor.OMM.Util;
using Sinacor.OMM.ConnectionManager.InternalQueue;
using Sinacor.OMM.Logging;
using PSE.Framework.Threading;

namespace ConnectionManagerTester
{
    public partial class formConnectionManagerTester : Form
    {
        const int KBYTE = 1024 * 1024;
        const int MEGABYTE = 1024 * 1024 * 1024;
        const string QUEUE_ID = "Minha Fila";

        MsmqSender _msmqSender;
        MsmqReceiver _msmqReceiver;

        InternalQueueManager<System.Messaging.Message> _iqManager;

        const string WRITE_QUEUE_ID = "Write";
        const string READ_QUEUE_ID = "Read";

        InternalQueueManager<BufferedMessage> _bufferedMessagesQueue = InternalQueueManager<BufferedMessage>.GetInstance();
//        InternalQueueManager<BufferedMessage> _bufferedSentMsgQueue = new InternalQueueManager<BufferedMessage>();

        int _messageSize;
        int _frequency;

        long _messagesSent = 0;
        long _messagesReceived = 0;
        long _messagesNoSent = 0;

        bool _keepSending;
        bool _keepWritingBuffer;

        WorkQueueType _workQueueTypeSet;

        public enum WorkQueueType
        {
            MSMQ,
            Internal
        }

        public formConnectionManagerTester()
        {
            InitializeComponent();
            _messageSize = int.Parse(numUmDownMessageSize.Value.ToString());
            _frequency = int.Parse(numUpDownFrequencia.Value.ToString());
        }


        private void btnStartSender_Click(object sender, EventArgs e)
        {
            btnStartSender.Enabled = false;

            if(_workQueueTypeSet == WorkQueueType.MSMQ)
                MsmqStartSender_Click();
            else
                IqStartSender_Click();

            btnStopSender.Enabled = true;
            btnSend.Enabled = true;
            btnStartLoopSending.Enabled = true;
            btnStopLoopSending.Enabled = false;
        }

        private void MsmqStartSender_Click()
        {
            _msmqSender = new MsmqSender(txtMsmqConnectionString.Text, int.Parse(numUpDownMessageLimit.Value.ToString()), long.Parse(numUpDownBytesLimit.Value.ToString()));
            _msmqSender.OnHealthStatusChanged += new MsmqSender.HealthStatusChanged(_msmqSender_OnHealthStatusChanged);
        }

        private void IqStartSender_Click()
        {
            _iqManager = InternalQueueManager<System.Messaging.Message>.GetInstance();
            _iqManager.OnHealthStatusChanged +=new InternalQueueManager<System.Messaging.Message>.HealthStatusChanged(_iqManager_OnHealthStatusChanged);
        }

        void  _iqManager_OnHealthStatusChanged(string queueId, PSE.Framework.Messaging.SynchronizedQueue<System.Messaging.Message> managedQueue)
        {
            SetText(txtMsgSntpMs, managedQueue.SentMessagesPerSecond.ToString());
            //SetText(txtSentQueueSize, managedQueue.QueueSizeBytes.ToString());
            SetText(txtSentQueuedMessages, managedQueue.Count.ToString());
        }

        private void btnStopSender_Click(object sender, EventArgs e)
        {
            _keepSending = false;
            btnSend.Enabled = false;
            btnStopSender.Enabled = false;
            btnStartLoopSending.Enabled = false;
            btnStopLoopSending.Enabled = false;
            
            if(_workQueueTypeSet == WorkQueueType.MSMQ)
            {
            if (_msmqSender != null)
                _msmqSender.Dispose();
            }
            btnStartSender.Enabled = true;
        }

        private void btnStartReceiver_Click(object sender, EventArgs e)
        {
            btnStartReceiver.Enabled = false;
            btnStopReceiver.Enabled = true;
            btnPauseReceiver.Enabled = true;
            btnResumeReceiver.Enabled = false;

            if (_workQueueTypeSet == WorkQueueType.MSMQ)
            {
                _msmqReceiver = new MsmqReceiver(txtMsmqConnectionString.Text);
                _msmqReceiver.MessageReceived += new MsmqReceiver.MessageArrival(_msmqReceiver_MessageReceived);
                _msmqReceiver.OnHealthStatusChanged += new MsmqReceiver.HealthStatusChanged(_msmqReceiver_OnHealthStatusChanged);
            }
            else
            {
                _iqManager = InternalQueueManager<System.Messaging.Message>.GetInstance();
                
                //_iqManager.SetMaxItensInBlockedEvent(QUEUE_ID, 2500);
                //_iqManager.SetMessageArrival(QUEUE_ID, IqMessageListArrival);
                _iqManager.SetMessageArrival(QUEUE_ID, IqMessageArrival);
                _iqManager.OnHealthStatusChanged +=new InternalQueueManager<System.Messaging.Message>.HealthStatusChanged(_iqManager_OnHealthStatusChangedRcv);
            }
        }

        void _msmqReceiver_MessageReceived(System.Messaging.Message msg, PSE.Framework.Messaging.MessageReceivedEventArgs e)
        {
            _messagesReceived++;

            BufferedWriteReceived((string)msg.Body, _messagesReceived);

            e.Result = true;
        }

        public void _iqManager_OnHealthStatusChangedRcv(string queueId, PSE.Framework.Messaging.SynchronizedQueue<System.Messaging.Message> sender)
        {
            SetText(txtMsgRcvdpMs, sender.ReceivedMessagesPerSecond.ToString());
            //SetText(txtReceivedQueueSize, sender.QueueSizeBytes.ToString());
            SetText(txtReceivedQueuedMessages, sender.Count.ToString());
        }

        private void IqMessageArrival(PSE.Framework.Messaging.SynchronizedQueue<System.Messaging.Message> sender, System.Messaging.Message item, string queueId)
        {
            _messagesReceived++;
            BufferedWriteReceived((string)item.Body, _messagesReceived);
        }        

        private void IqMessageListArrival(List<System.Messaging.Message> listItens, string queueId)
        {
            foreach (System.Messaging.Message iten in listItens)
            {
                _messagesReceived++;
                BufferedWriteReceived((string)iten.Body, _messagesReceived);
            }
        }

        void _msmqReceiver_OnHealthStatusChanged(MsmqReceiver msmqReceiver)
        {
            SetText(txtMsgRcvdpMs, msmqReceiver.MessagesReceivePerSecond.ToString());
            SetText(txtReceivedQueueSize, msmqReceiver.QueueSizeBytes.ToString());
            SetText(txtReceivedQueuedMessages, msmqReceiver.QueuedMessages.ToString());
        }

        private void BufferedWriteReceived(string message, long messagesReceived)
        {
            BufferedMessage bufMsg = new BufferedMessage();
            bufMsg.Message = message;
            bufMsg.MessageNumer = messagesReceived;

            lock (_bufferedMessagesQueue)
                _bufferedMessagesQueue.Enqueue(READ_QUEUE_ID, bufMsg);
        }

        private void BufferedWriteSent(string message, long messagesReceived)
        {
            BufferedMessage bufMsg = new BufferedMessage();
            bufMsg.Message = message;
            bufMsg.MessageNumer = messagesReceived;

            _bufferedMessagesQueue.Enqueue(WRITE_QUEUE_ID, bufMsg);
        }

        private void WriteBufferedMessages()
        {
            BufferedMessage bufMsg;

            _keepWritingBuffer = true;
            while (IsWritingBuffer())
            {

                if (_bufferedMessagesQueue.Count(READ_QUEUE_ID) > 0)
                {
                    while (_bufferedMessagesQueue.Count(READ_QUEUE_ID) > 0)
                    {
                        bufMsg = _bufferedMessagesQueue.Dequeue(READ_QUEUE_ID);
                        if (_bufferedMessagesQueue.Count(READ_QUEUE_ID) == 0)
                        {
                            SetText(txtMsgReceived, bufMsg.Message);
                            SetText(txtNumMsgsReceived, bufMsg.MessageNumer.ToString());
                        }
                    }
                }

                if (_bufferedMessagesQueue.Count(WRITE_QUEUE_ID) > 0)
                {
                    while (_bufferedMessagesQueue.Count(WRITE_QUEUE_ID) > 0)
                    {
                        bufMsg = _bufferedMessagesQueue.Dequeue(WRITE_QUEUE_ID);
                        if (_bufferedMessagesQueue.Count(WRITE_QUEUE_ID) == 0)
                        {
                            SetText(txtNumMsgsSent, _messagesSent.ToString());
                        }
                    }
                }

                Sleeper.Sleep(750, IsWritingBuffer);
            }
        }

        private bool IsWritingBuffer()
        {
            return _keepWritingBuffer;
        }

        private void btnStopReceiver_Click(object sender, EventArgs e)
        {
            btnStopReceiver.Enabled = false;
            btnPauseReceiver.Enabled = false;
            btnResumeReceiver.Enabled = false;

            if (_workQueueTypeSet == WorkQueueType.MSMQ)
            {
                if (_msmqReceiver != null)
                {
                    _msmqReceiver.Dispose();
                }
            }
            else
            {
                _iqManager.Dispose();
            }

            btnStartReceiver.Enabled = true;
        }

        private delegate void dSetText(TextBox textBox,  string texto);
        private void SetText(TextBox textBox,  string texto)
        {
            if (textBox.InvokeRequired)
            {
                try
                {
                    dSetText d = new dSetText(SetText);
                    this.Invoke(d, new object[] { textBox, texto });
                }
                catch (Exception) { }
            }
            else
                textBox.Text = texto;
        }

        private void btnSend_Click(object sender, EventArgs e)
        {
            System.Messaging.Message msg = new System.Messaging.Message();
            msg.Body = txtMessageToSend.Text;

            if (_workQueueTypeSet == WorkQueueType.MSMQ)
            {
                if (_msmqSender.Send(msg))
                    _messagesSent++;
                else
                    _messagesNoSent++;
            }
            else
            {
                _iqManager.Enqueue(QUEUE_ID, msg);
                _messagesSent++;
            }
            
            txtNumMsgsSent.Text = _messagesSent.ToString();
            txtNumNoSentMsg.Text = _messagesNoSent.ToString();

        }

        private void btnStartLoopSending_Click(object sender, EventArgs e)
        {
            btnStartLoopSending.Enabled = false;
            
            StartLoopSending();

            btnStopLoopSending.Enabled = true;
        }

        private void StartLoopSending()
        {
            Thread thr = new Thread(new ThreadStart(SendMessageUntilStop));
            thr.Start();
        }

        private void numUmDownMessageSize_ValueChanged(object sender, EventArgs e)
        {
            _messageSize = int.Parse(numUmDownMessageSize.Value.ToString());
            DefineMsg();
        }

        private void numUpDownFrequencia_ValueChanged(object sender, EventArgs e)
        {
            _frequency = int.Parse(numUpDownFrequencia.Value.ToString());
        }

        private void DefineMsg()
        {
            StringBuilder sb = new StringBuilder("");
            _message = sb.Append('*', _messageSize).ToString();
        }

        string _message;
        public void SendMessageUntilStop()
        {
            System.Messaging.Message msg = new System.Messaging.Message();
            
            DefineMsg();

            _keepSending = true;

            while (_keepSending)
            {
                
                msg.Body = _messagesSent.ToString() + "-" + _message;

                if (_workQueueTypeSet == WorkQueueType.MSMQ)
                {
                    if (_msmqSender.Send(msg))
                        _messagesSent++;
                    else
                    {
                        _messagesNoSent++;
                        SetText(txtNumNoSentMsg, _messagesNoSent.ToString());
                    }
                }
                else
                {
                    _iqManager.Enqueue(QUEUE_ID, msg);
                    _messagesSent++;
                }

                BufferedWriteSent(null, _messagesSent);

                if(_frequency>0)
                    Thread.Sleep(_frequency);
                
            }
        }

        private void btnStopLoopSending_Click(object sender, EventArgs e)
        {
            btnStopLoopSending.Enabled = false;
            _keepSending = false;
            btnStartLoopSending.Enabled = true;
        }

        void _msmqSender_OnHealthStatusChanged(MsmqSender msmqSender)
        {
            SetText(txtMsgSntpMs, msmqSender.MessagesSentPerSecond.ToString());
            SetText(txtSentQueueSize, msmqSender.QueueSizeBytes.ToString());
            SetText(txtSentQueuedMessages, msmqSender.QueuedMessages.ToString());
        }

        private bool IsSending()
        {
            return _keepSending;
        }

        private void btnSleep_Click(object sender, EventArgs e)
        {
            Thread thr1 = new Thread(new ThreadStart(tstSleepOnSend));
            Thread thr2 = new Thread(new ThreadStart(tstSleepOnReceive));
            thr1.Start();
            thr2.Start();
        }

        private void tstSleepOnSend()
        {
            for (int x = 0; x < 100; x++)
            {
                Sleeper.Sleep(1, IsSending);
                SetText(txtMessageToSend, txtMessageToSend.Text + "\r\n" + DateTime.Now.ToString("hh:mm:ss.fff"));
            }
        }

        private void tstSleepOnReceive()
        {
            for (int x = 0; x < 100; x++)
            {
                Sleeper.Sleep(5, IsSending);
                SetText(txtMsgReceived, txtMsgReceived.Text + "\r\n" + DateTime.Now.ToString("hh:mm:ss.fff"));
            }
        }

        private void numUpDownBytesLimit_ValueChanged(object sender, EventArgs e)
        {
            lblLimitMb.Text = Math.Round((numUpDownBytesLimit.Value / 1024 / 1024 / 1024),0).ToString() + " Mb";

            if (_msmqSender != null)
                _msmqSender.QueuedBytesLimit = long.Parse(Math.Round(numUpDownBytesLimit.Value, 0).ToString());
        }

        private class BufferedMessage
        {
            public string Message { get; set; }
            public long MessageNumer { get; set; }
        }

        private void numUpDownMessageLimit_ValueChanged(object sender, EventArgs e)
        {
            if(_msmqSender != null)
            _msmqSender.QueuedMessageLimit = int.Parse(Math.Round(numUpDownMessageLimit.Value, 0).ToString());
        }

        private void formConnectionManagerTester_FormClosed(object sender, FormClosedEventArgs e)
        {
            _keepWritingBuffer = false;
            btnStopLoopSending_Click(sender, e);
            btnStopSender_Click(sender, e);
            btnStopReceiver_Click(sender, e);
            if (_workQueueTypeSet == WorkQueueType.MSMQ)
            {
                if (_msmqReceiver != null)
                    _msmqReceiver.Dispose();

                if (_msmqSender != null)
                    _msmqSender.Dispose();
            }
            else
            {
                _iqManager.Dispose();
            }

            Application.Exit();
        }

        private void btnPauseReceiver_Click(object sender, EventArgs e)
        {
            btnPauseReceiver.Enabled = false;
            btnResumeReceiver.Enabled = true;
            if(_workQueueTypeSet == WorkQueueType.MSMQ)
            {
                _msmqReceiver.PauseMsmqReceiver();
            }
            else
            {
                _iqManager.PauseReceiver(QUEUE_ID);
            }
        }

        private void btnResumeReceiver_Click(object sender, EventArgs e)
        {
            btnResumeReceiver.Enabled = false;
            btnPauseReceiver.Enabled = true;
            if (_workQueueTypeSet == WorkQueueType.MSMQ)
            {
                _msmqReceiver.ResumeMsmqReceiver();
            }
            else
                _iqManager.ResumeReceiver(QUEUE_ID);
        }

        private void txtSentQueueSize_TextChanged(object sender, EventArgs e)
        {
            decimal size;

            size = decimal.Parse(txtSentQueueSize.Text);

            if (size < KBYTE)
                lblSentQueueSize.Text = "< 1 kb";
            else if(size < MEGABYTE)
                lblSentQueueSize.Text = Math.Round((size / KBYTE), 2).ToString() + " Kb";
            else
                lblSentQueueSize.Text = Math.Round((size / MEGABYTE), 2).ToString() + " Mb";

        }

        private void formConnectionManagerTester_Load(object sender, EventArgs e)
        {
            Thread thr = new Thread(new ThreadStart(WriteBufferedMessages));
            thr.Start();
        }

        private void formConnectionManagerTester_FormClosing(object sender, FormClosingEventArgs e)
        {
            formConnectionManagerTester_FormClosed(sender, new FormClosedEventArgs(CloseReason.UserClosing));
        }

        private void txtReceivedQueueSize_TextChanged(object sender, EventArgs e)
        {
            decimal size;

            size = decimal.Parse(txtReceivedQueueSize.Text);

            if (size < KBYTE)
                lblReceivedQueueSize.Text = "< 1 Kb";
            else if (size < MEGABYTE)
                lblReceivedQueueSize.Text = Math.Round((size / KBYTE), 2).ToString() + " Kb";
            else
                lblReceivedQueueSize.Text = Math.Round((size / MEGABYTE), 2).ToString() + " Mb";

        }

        private void RadioInternalQueue_CheckedChanged(object sender, EventArgs e)
        {
            if (((RadioButton)sender).Text == "InternalQueue")
                txtMsmqConnectionString.Enabled = false;

        }

        private void radioMSMQ_CheckedChanged(object sender, EventArgs e)
        {
            if (((RadioButton)sender).Checked)
            {
                if (((RadioButton)sender).Text == "InternalQueue")
                    radioMSMQ.Checked = false;
                else
                    RadioInternalQueue.Checked = false;

            }

            if (((RadioButton)sender).Text == "MSMQ")
                txtMsmqConnectionString.Enabled = true;

        }

        private void btnOK_Click(object sender, EventArgs e)
        {
            groupSetQueueType.Enabled = false;
            groupCtrlApp.Enabled = true;

            if (radioMSMQ.Checked)
                _workQueueTypeSet = WorkQueueType.MSMQ;
            else
                _workQueueTypeSet = WorkQueueType.Internal;
        }













    }
}
